package com.sheep.util;

import com.sheep.api.bean.TreeNodeBean;
import com.sheep.api.bean.UserBean;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.log4j.Logger;

import java.io.*;
import java.net.URI;

public class HDFSFileSupport {
    private static final Logger logger = Logger.getLogger(HDFSFileSupport.class);
    public static UserBean user = new UserBean();

    public static void main(String[] args) {
        HDFSFileSupport uploadAndDown = new HDFSFileSupport();
        try {
            System.setProperty("hadoop.home.dir", "D:\\hadoop-2.6.4");
            System.setProperty("HADOOP_USER_NAME", "root");
            String dst = ConfSupport.getUserHDFSAddress(new UserBean()) + "/b-2016-04-14.json";
            String local = "D:\\a.json";
            long startT = System.currentTimeMillis();
            //将本地文件local.txt上传为HDFS上cloud.txt文件
            uploadAndDown.upLoadToCloud(local, dst);
//            String lo =ConfSupport.getDefaultAddress() + "/ADSB1/2016/04/06/B-2358#MU2060-m-00000";
//            String shor = ConfSupport.getDefaultAddress()+"/ADSB1/2016/04/06/B-1873#CA8232-m-00000";
//            uploadAndDown.downFromCloud("d:\\local.json", lo);
//            uploadAndDown.downFromCloud("d:\\local.json",shor);
//            System.out.println(uploadAndDown.getCloudFileByString(lo));
            logger.info("upload to cloud:" + (System.currentTimeMillis() - startT));
            //将HDFS上的cloud.txt文件下载到本地cloudTolocal.txt文件
//            uploadAndDown.downFromCloud(local, dst);
//            FileSystem hdfs = FileSystem.get(URI.create(dst),ConfSupport.getConf());
//            FileStatus[] fs = hdfs.listStatus(new Path(dst));
//            Path[] listPath = FileUtil.stat2Paths(fs);
//            TreeNodeBean top = new TreeNodeBean();
//            top.setPath(dst);
//            top = convert2Tree(top);
//            System.out.println(top.show());
//            for(Path p : listPath){
//                System.out.println(p);
//                FileSystem tempfs = p.getFileSystem(ConfSupport.getConf());
//                FileStatus[] tempfst = tempfs.listStatus(p);
//                Path[] temppaths = FileUtil.stat2Paths(tempfst);
//                for(Path j:temppaths){
//                    System.out.println("    "+j);
//                }
//            }
//            FileSystem fs = FileSystem.get(URI.create(dst), ConfSupport.getConf());
//            System.out.println(fs.getHomeDirectory().getName());
//            for(FileSystem i:fs.getChildFileSystems()){
//                System.out.println(i.getUri().getPath());
//            }
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    public static TreeNodeBean convert2Tree(TreeNodeBean parent) throws Exception {
        FileSystem hdfs = FileSystem.get(URI.create(parent.getPath()), ConfSupport.getConf());
        FileStatus[] fs = hdfs.listStatus(new Path(parent.getPath()));
        for (int i = 0; i < fs.length; i++) {
            Path p = fs[i].getPath();
            TreeNodeBean me = new TreeNodeBean();
            me.setPath(p.toString());
            me.setParent(parent);
            me.setIsDir(fs[i].isDirectory());
            if (me.isDir()) {
                parent.getChildren().add(convert2Tree(me));
            } else {
                parent.getChildren().add(me);
            }
        }
        return parent;
    }

    public static boolean isExist(String dst) {
        try {
            FileSystem hdfs = null;
            hdfs = FileSystem.get(URI.create(dst), ConfSupport.getConf());
            FileStatus[] fs = hdfs.listStatus(new Path(dst));
        } catch (IOException e) {
            logger.error("目录或文件不存在:" + e.getMessage());
            return false;
        }
        return true;
    }

    /**
     * LOCAL_SRC,本地文件存取的位置
     * CLOUD_DEST,存放到云端HDFS的位置
     */
    public static void upLoadToCloud(String LOCAL_SRC, String CLOUD_DEST)
            throws IOException {
        InputStream in = new BufferedInputStream(new FileInputStream(LOCAL_SRC));
        // 文件系统
        FileSystem fs = FileSystem.get(URI.create(CLOUD_DEST), ConfSupport.getConf());
        // 输出流
        OutputStream out = fs.create(new Path(CLOUD_DEST)
//                , new Progressable() {
//            public void progress() {
//                System.out.println("上传完成一个文件到HDFS"+(a++));
//            }
//        }
        );
        // 连接两个流，形成通道，使输入流向输出流传输数据
        IOUtils.copyBytes(in, out, 1024, true);
    }

    public static String getCloudFileByString(String CLOUD_DESC) throws IOException {
        // 实例化一个文件系统
        FileSystem fs = FileSystem.get(URI.create(CLOUD_DESC), ConfSupport.getConf());
        // 读出流
        FSDataInputStream HDFS_IN = fs.open(new Path(CLOUD_DESC));
        // 写入流
        OutputStream strOUT = new ByteArrayOutputStream();
        // 将InputStrteam 中的内容通过IOUtils的copyBytes方法复制到OutToLOCAL中
        IOUtils.copyBytes(HDFS_IN, strOUT, 1024, true);
        return strOUT.toString();
    }

    /**
     * LOCAL_SRC,down到本地的文件
     * CLOUD_DEST,云端HDFS上的文件
     */
    public static void downFromCloud(String LOCAL_SRC, String CLOUD_DESC) throws IOException {
        // 实例化一个文件系统
        FileSystem fs = FileSystem.get(URI.create(CLOUD_DESC), ConfSupport.getConf());
        // 读出流
        FSDataInputStream HDFS_IN = fs.open(new Path(CLOUD_DESC));
        // 写入流
        OutputStream OutToLOCAL = new FileOutputStream(LOCAL_SRC);
        // 将InputStrteam 中的内容通过IOUtils的copyBytes方法复制到OutToLOCAL中
        IOUtils.copyBytes(HDFS_IN, OutToLOCAL, 1024, true);
    }

    public static void renameFile(String str, String dst) {
        FileSystem fs = null;
        try {
            fs = FileSystem.get(ConfSupport.getConf());
            Path srcPath = new Path(str);
            Path dstPath = new Path(dst);
            fs.rename(srcPath, dstPath);
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            if (fs != null) {
                try {
                    fs.close();
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        }
    }

    public static void deleteFile(String src) {
        FileSystem fs = null;
        try {
            fs = FileSystem.get(ConfSupport.getConf());
            fs.delete(new Path(src));
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            if (fs != null) {
                try {
                    fs.close();
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        }
    }

}
